import datetime
from airflow.providers.dingding.operators.dingding import DingdingHook
from airflow.providers.presto.hooks.presto import PrestoHook
from utils.macros import cst_ds_sfm, cst_ds_nodash
from airflow.hooks.postgres_hook import PostgresHook

def subtime(date1,date2):
    date1=datetime.datetime.strptime(date1, "%Y-%m-%d %H:%M:%S")
    date2=datetime.datetime.strptime(date2, "%Y-%m-%d %H:%M:%S")

    return date2-date1

def yl_threeSegCodeOnFailure(kwargs,dingding_conn_id="dingding_ThreeSeg_etl_alert", at_mobiles=None,at_all=None):
    def  _yl_threeSegCodeOnFailure(ctx):
        #envInfo 环境基本信息
        currentEndTime = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        rundt = datetime.datetime.now().strftime("%Y-%m-%d")
        current_dag = ctx["ti"]
        env = ctx['var']['value'].env
        execution_date = cst_ds_nodash(ctx["execution_date"])

        #监控表， 参数传参，库名，表名，表描述，表唯一id，数据是否隐私【1：隐私，0：不隐私 】
        db = kwargs["db"]
        table = kwargs["table"]
        desc = kwargs["desc"]
        taskid = kwargs["taskid"]
        ifprivacy = kwargs["ifprivacy"]

        # 使用presto计算 监控表日数据量
        try:
            presql = f"select count(1) as cnt from {db}.{table} where dt='{execution_date}'"
            preconn = PrestoHook(presto_conn_id='presto_default')
            hivecnt = preconn.get_first(presql)[0]
        except Exception as err:
            print(err)
            hivecnt = 0


        # 任务失败状态下，下列指标都置为0
        count_max_30 = 0
        count_min_30 = 0
        count_avg_30 = 0
        count_avg_7 = 0

        # 需要写入postgres 的日志数据
        taskid,dbname,tablename,starttime,endtime,elpasetime,executiondate,count_daily = \
            taskid, db, table, cst_ds_sfm(current_dag.start_date), currentEndTime,subtime(cst_ds_sfm(current_dag.start_date), currentEndTime), execution_date, hivecnt
        count_max_30,count_min_30,count_avg_30,count_avg_7,avg_30_ratio,avg_7_ratio, isok, rundt = \
            count_max_30, count_min_30, count_avg_30, count_avg_7, 0, 0, current_dag.state, rundt

        # 往 postgre 写入日志
        cols_list = ["taskid", "dbname", "tablename", "starttime", "endtime", "elpasetime", "executiondate", "count_daily", "count_max_30", "count_min_30", "count_avg_30", "count_avg_7", "avg_30_ratio", "avg_7_ratio", "isok", "rundt"]
        values_str = (taskid, dbname, tablename, starttime, endtime, elpasetime, executiondate, count_daily, count_max_30, count_min_30, count_avg_30, count_avg_7, avg_30_ratio, avg_7_ratio, isok, rundt)

        #posconn = PostgresHook(postgres_conn_id='postgres_last')
        posconn = PostgresHook(postgres_conn_id='postgresql_pro')
        posconn.insert_rows("public.ai_offline_etl_log", [values_str], cols_list)

        # 数据隐私性处理
        if (ifprivacy):
            dailycount = "******"
        else:
            dailycount = count_daily
        # 数据异常，切换告警机器人发送到异常告警群，否则发送正常日志群

        title = f'[{env.upper()}]: {current_dag.task_id.upper()} 任务失败告警'

        # 发送钉钉的任务日志
        text = f"""
# # <font color=#EA3323>{title}</font>
   - *表全名:* {dbname}.{tablename}
   - *任务状态:* {isok}
   - *分区:* {executiondate}
   - *分区数据量:* {dailycount}
   - *7天均值同环比:* 0%
   - *30天均值同环比:* 0%
   - *任务开始时间:* {starttime}
   - *任务结束时间:* {currentEndTime}
   - *任务耗时:* {elpasetime}
   - *表描述:* {desc}
"""

        #dingding_conn_id = "dingding_last"
        return DingdingHook(dingding_conn_id=dingding_conn_id,
                            message_type='markdown',  # text, link, markdown, actionCard, feedCard
                            message={'title': title, 'text': text},
                            at_mobiles=at_mobiles,
                            at_all=at_all, ).send()

    return _yl_threeSegCodeOnFailure